package com.itheima.stream;

import cn.hutool.core.convert.Convert;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;

import java.time.Duration;
import java.util.Arrays;

//@Configuration
//@Slf4j
public class WordCountsDemo5 {

    /*
    *
    * TopicA 15秒内的wordcount，结果写入TopicB
    比如登录某app，20分钟内不操作，会自动退出。
    一个典型的案例是，希望通过Session Window计算某个用户访问网站的时间。
    对于一个特定的用户（用Key表示）而言，当发生登录操作时，该用户（Key）的窗口即开始，当发生退出操作或者超时时，
    该用户（Key）的窗口即结束。窗口结束时，可计算该用户的访问时间或者点击次数等。
    * */

    @Bean
    public KTable<Windowed<Object>, Long> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream("itcast-topic-input");
        KTable<Windowed<Object>, Long> countKtable = stream.flatMapValues(new ValueMapper<String, Iterable<?>>() {
            @Override
            public Iterable<?> apply(String value) {
                return Arrays.asList(value.toString().split(" "));
            }
        }).map(new KeyValueMapper<String, Object, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(String key, Object value) {
                return new KeyValue<String, String>(Convert.toStr(value), "1");
            }
        }).groupByKey()
                //.windowedBy(TimeWindows.of(Duration.ofSeconds(25).toMillis()).advanceBy(Duration.ofSeconds(15).toMillis()))
                //.windowedBy(TimeWindows.of(Duration.ofSeconds(15).toMillis()))
                .windowedBy(SessionWindows.with(Duration.ofSeconds(15).toMillis()))
                .count();
        countKtable.toStream().map(new KeyValueMapper<Windowed<Object>, Long, KeyValue<?, ?>>() {
            @Override
            public KeyValue<?, ?> apply(Windowed<Object> key, Long value) {
                return new KeyValue<String, String>(key.toString(), value.toString());
            }
        }).to("itcast-topic-out");

        return countKtable;
    }
}